package com.mayikt.consumer;

import com.alibaba.fastjson.JSONObject;

import com.mayikt.entity.OrderEntity;
import com.mayikt.entity.SeckillEntity;
import com.mayikt.mapper.OrderMapper;
import com.mayikt.mapper.SeckillMapper;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Component
public class InventoryConsumer {
    @Autowired
    private SeckillMapper seckillMapper;
    @Autowired
    private OrderMapper orderMapper;

    @KafkaListener(topics = "mayikt-topic-spike")
    @Transactional
    public void receive(ConsumerRecord<?, ?> consumer) throws Exception {
        String json = (String) consumer.value();
        if (StringUtils.isEmpty(json)) {
            return;
        }
        JSONObject jsonObject = JSONObject.parseObject(json);
        String userPhone = jsonObject.getString("userPhone");
        Long seckillId = jsonObject.getLong("seckillId");
        // 根据userid+seckillId+msg自带的消息全局id 解决幂等性问题
        //  3.根据该id查询该商品是否存在
        SeckillEntity seckillEntity = seckillMapper.findBySeckillId(seckillId);
        if (seckillEntity == null) {
            return;
        }
        // 4.对库存实现减去1
        Long version = seckillEntity.getVersion();
        int resultSeckill = seckillMapper.inventoryDeduction(seckillId, version);
        if (resultSeckill <= 0) {
            return;
        }
        // 5.插入订单记录
        OrderEntity orderEntity = new OrderEntity(seckillId, userPhone);
        int resultOrder = orderMapper.insertOrder(orderEntity);
        if (resultOrder <= 0) {
            return;
        }
    }
}
